package com.jaymin.flink.datastream.sink;

import com.google.common.base.Splitter;
import com.google.common.collect.Lists;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSink;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import java.math.BigDecimal;
import java.util.List;

/**
 * @author jaymin
 * @since 2024/4/5 18:50
 */
public class SinkApp {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment executionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<String> streamSource = executionEnvironment.readTextFile("data/access.log");
        streamSource.map(new MapFunction<String, Tuple2<String, BigDecimal>>() {
                    @Override
                    public Tuple2<String, BigDecimal> map(String s) throws Exception {
                        String[] splits = s.split(",");
                        String domain = splits[1].trim();
                        BigDecimal traffic = new BigDecimal(splits[2].trim());
                        return Tuple2.of(domain, traffic);
                    }
                })
                .keyBy(t -> t.f0)
                .reduce(new ReduceFunction<Tuple2<String, BigDecimal>>() {
                    @Override
                    public Tuple2<String, BigDecimal> reduce(Tuple2<String, BigDecimal> a1, Tuple2<String, BigDecimal> a2) throws Exception {
                        return Tuple2.of(a1.f0, a1.f1.add(a2.f1));
                    }
                })
                .addSink(new PKMysqlSink());
        executionEnvironment.execute("Sink App");
    }
}
